package com.yet.center;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;

import com.yet.exception.ConfigException;
import com.yet.exception.ConfigStartException;
import com.yet.listener.YetListener;
import com.yet.listener.YetPathEvent;
import com.yet.serialization.Serialization;

public class ZookeeperConfigCenter extends ConfigCenter implements Container {

	public static final ConcurrentMap<String, PathChildrenCache> childrens = new ConcurrentHashMap<>();

	private final CuratorFramework client;

	private final Serialization serialization;

	public ZookeeperConfigCenter(CuratorFramework client, Serialization serialization) {
		super();
		this.client = client;
		this.serialization = serialization;
	}

	@SuppressWarnings("unchecked")
	@Override
	public <T> T getValue(String path, Class<T> clz) throws ConfigException {
		try {
			return (T) serialization.deserialize(client.getData().forPath(getWholePath(path)), clz);
		} catch (Exception e) {
			throw new ConfigException(e);
		}
	}

	@Override
	public Object getValue(String path) throws ConfigException {
		try {
			return client.getData().forPath(getWholePath(path));
		} catch (Exception e) {
			throw new ConfigException(e);
		}
	}

	@Override
	public void subscribe(String path, YetListener listener) throws ConfigException {
		try {
			boolean isProfile = StringUtils.isBlank(path);
			if (isProfile) {
				path = getProfilePath();
			}
			PathChildrenCache children = childrens.get(path);
			if (children == null) {
				children = new PathChildrenCache(client, getWholePath(isProfile ? null : path), true);
				children.start();
				childrens.put(path, children);
			}
			children.getListenable().addListener((client, event) -> listener
					.onApplicationEvent(new YetPathEvent(event.getType().toString(), event.getData())));
		} catch (Exception e) {
			throw new ConfigException(e);
		}
	}

	@Override
	public void unSubscribe(String path) throws ConfigException {
		Optional.ofNullable(childrens.get(path)).ifPresent(c -> c.getListenable().clear());
	}

	@Override
	public void start() throws ConfigStartException {
		client.start();
		try {
			createNode(ROOT);
			createNode(getProfilePath());
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void stop() {
		client.close();
	}

	@Override
	public void setValue(String path, Object data) throws ConfigException {
		try {
			setData(getWholePath(path), data);
		} catch (Exception e) {
			throw new ConfigException(e);
		}
	}

	private void setData(String path, Object data) throws Exception {
		if (client.checkExists().forPath(path) == null) {
			client.create().forPath(path, serialization.serialize(data));
			return;
		}
		client.setData().forPath(path, serialization.serialize(data));
	}

	private void createNode(String path) throws Exception {
		if (client.checkExists().forPath(path) == null) {
			client.create().forPath(path);
		}
	}

	@Override
	public List<String> getRoot() throws ConfigException {
		try {
			return client.getChildren().forPath(getProfilePath());
		} catch (Exception e) {
			throw new ConfigException(e);
		}
	}

	@Override
	public Map<String, Object> load() throws ConfigException {
		try {
			return client.getChildren().forPath(getProfilePath()).stream()
					.collect(Collectors.toMap(k -> k, v -> getValue(v, String.class)));
		} catch (Exception e) {
			throw new ConfigException(e);
		}
	}
}
